package cn.nhs.schedule.service;

import cn.nhs.schedule.common.Constants;
import cn.nhs.schedule.domain.Instruct;
import cn.nhs.schedule.task.CronTaskRegister;
import cn.nhs.schedule.task.SchedulingRunnable;
import cn.nhs.schedule.util.StrUtil;
import com.alibaba.fastjson.JSON;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.nodes.PersistentEphemeralNode;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import static cn.nhs.schedule.common.Constants.Global.*;
/**
 封装用于与Zookeeper交互的工具类.它提供了一些方法来创建、删除、设置数据以及监听ZooKeeper节点的变化。
 */
public class ZkCuratorServer {

    // 日志服务
    private static Logger logger = LoggerFactory.getLogger(ZkCuratorServer.class);

    // 获取zk的连接
    public static CuratorFramework getClient(String connectString) {
        // 检查全局变量 Constants.Global.client 是否已经存在实例，如果存在，则直接返回该实例，避免重复创建。
        if (null != Constants.Global.client) return Constants.Global.client;

        // 如果 Constants.Global.client 为空，表示还没有创建过实例，则使用给定的 connectString
        // 和重试策略 ExponentialBackoffRetry 创建一个新的 CuratorFramework 实例。
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(connectString, retryPolicy);

        //添加重连监听
        // 添加一个连接状态的监听器，当连接状态发生变化时，
        // 会执行对应的回调函数。在这段代码中，只处理了连接成功和重新连接成功的情况，分别打印日志信息。
        client.getConnectionStateListenable().addListener((curatorFramework, connectionState) -> {
            switch (connectionState) {
                case CONNECTED:
                    logger.info("schedule init server connected {}", connectString);
                    break;

                case RECONNECTED:

                    break;
                default:
                    break;
            }
        });
        client.start();
        Constants.Global.client = client;
        return client;
    }

    //所有子节点监听
    public static void addTreeCacheListener(final ApplicationContext applicationContext, final CuratorFramework client, String path) throws Exception {
        /**
         具体而言，TreeCache是ZooKeeper的一个监听器，可以监控指定节点及其子节点的变化。
         通过创建TreeCache实例并启动它，可以在ZooKeeper中的指定节点上设置监听器，以便在节点发生变化时触发相应的事件。
         启动TreeCache后，它会从指定节点开始递归地缓存其下所有的子节点和数据，并且持续监控这些节点的状态变化。
         */
        TreeCache treeCache = new TreeCache(client, path);
        treeCache.start();
        // 为 treeCache 添加一个监听器，当节点发生变化时，会调用对应的回调函数进行处理。
        treeCache.getListenable().addListener((curatorFramework, event) -> {

            // 这段代码的作用是从 ZooKeeper 的监听事件中解析出一个 Instruct 对象。
            /*
            具体来说，代码首先判断事件中是否包含有效数据，如果没有则直接返回。然后将事件中的数据转换为字节数组，
            再将字节数组转换为字符串，并根据一些条件进行判断，确保该字符串是一个合法的 JSON 格式。

            如果判断失败，则直接返回。若判断成功，则利用 JSON.parseObject() 方法将该 JSON
            字符串解析成一个 Instruct 对象，并返回该对象。
            */
            if (null == event.getData()) return;
            byte[] eventData = event.getData().getData();
            if (null == eventData || eventData.length < 1) return;
            String json = new String(eventData, Constants.Global.CHARSET_NAME);
            if ("".equals(json) || json.indexOf("{") != 0 || json.lastIndexOf("}") + 1 != json.length()) return;
            Instruct instruct = JSON.parseObject(new String(event.getData().getData(), Constants.Global.CHARSET_NAME), Instruct.class);
            // 在回调函数中，判断事件的类型，如果是节点新增或更新，则根据节点中存储的信息以及一些条件进行相应的业务逻辑处理。
            switch (event.getType()) {
                case NODE_ADDED:
                case NODE_UPDATED:
                    // 如果当前本机的ip 与 schedulerServerId 都与 回调返回的相等。
                    if (Constants.Global.ip.equals(instruct.getIp()) &&
                            Constants.Global.schedulerServerId.equals(instruct.getSchedulerServerId())) {

                        //获取对象
                        CronTaskRegister cronTaskRegistrar = applicationContext.getBean("nhs-schedule-cronTaskRegister", CronTaskRegister.class);
                        boolean isExist = applicationContext.containsBean(instruct.getBeanName());
                        if (!isExist) return;
                        Object scheduleBean = applicationContext.getBean(instruct.getBeanName());
                        // /cn/nhs/schedule/server/schedule-spring-boot-starter-test/ip/机器ip/clazz/类对象名称/method/方法名称/status
                        String path_root_server_ip_clazz_method_status = StrUtil.joinStr(path_root, Constants.Global.LINE, "server", Constants.Global.LINE, instruct.getSchedulerServerId(), Constants.Global.LINE, "ip", LINE, instruct.getIp(), LINE, "clazz", LINE, instruct.getBeanName(), LINE, "method", LINE, instruct.getMethodName(), "/status");
                        //执行命令 0关闭、1启动、2更新
                        Integer status = instruct.getStatus();
                        switch (status) {
                            case 0: // 关闭
                                cronTaskRegistrar.removeCronTask(instruct.getBeanName() + "_" + instruct.getMethodName());
                                // 重新将状态设置回去
                                setData(client, path_root_server_ip_clazz_method_status, "0");
                                logger.info("schedule task stop {} {}", instruct.getBeanName(), instruct.getMethodName());
                                break;
                            case 1: // 启动

                                cronTaskRegistrar.addCronTask(new SchedulingRunnable(scheduleBean, instruct.getBeanName(), instruct.getMethodName()), instruct.getCron());
                                setData(client, path_root_server_ip_clazz_method_status, "1");
                                logger.info("schedule task start {} {}", instruct.getBeanName(), instruct.getMethodName());
                                break;
                            case 2: // 更新
                                cronTaskRegistrar.removeCronTask(instruct.getBeanName() + "_" + instruct.getMethodName());
                                cronTaskRegistrar.addCronTask(new SchedulingRunnable(scheduleBean, instruct.getBeanName(), instruct.getMethodName()), instruct.getCron());
                                setData(client, path_root_server_ip_clazz_method_status, "1");
                                logger.info("schedule task refresh {} {}", instruct.getBeanName(), instruct.getMethodName());
                                break;
                        }
                    }
                    break;
                case NODE_REMOVED:
                    break;
                default:
                    break;
            }
        });
    }

    //设置数据
    public static void setData(CuratorFramework client, String path, String data) throws Exception {
        if(null == client.checkExists().forPath(path)) return;
        client.setData().forPath(path, data.getBytes(Constants.Global.CHARSET_NAME));
    }



    // 递归删除节点
    public static void deletingChildrenIfNeeded(CuratorFramework client, String path) throws Exception {
        if (null == client.checkExists().forPath(path)) return;
        // 递归删除节点
        client.delete().deletingChildrenIfNeeded().forPath(path);
    }

    //创建节点

    /**
     这段代码的作用是根据给定的路径逐级创建ZooKeeper节点，确保指定路径下的所有父节点都存在，如果不存在则创建。

     具体来说，代码首先创建一个空的字符串列表pathChild，然后向其中添加初始路径path。接下来使用while循环，
     每次循环找到path中最后一个Constants.Global.LINE（可能是文件目录分隔符）的位置，
     然后将path截取为该位置之前的子串，并将该子串添加到pathChild中。这样就将path拆分成了各级父路径并存储在pathChild中。
     */
    public static void createNode(CuratorFramework client, String path) throws Exception {
        List<String> pathChild = new ArrayList<>();
        pathChild.add(path);
        while (path.lastIndexOf(Constants.Global.LINE) > 0) {
            path = path.substring(0, path.lastIndexOf(Constants.Global.LINE));
            pathChild.add(path);
        }
        for (int i = pathChild.size() - 1; i >= 0; i--) {
            Stat stat = client.checkExists().forPath(pathChild.get(i));
            if (null == stat) {
                client.create().creatingParentsIfNeeded().forPath(pathChild.get(i));
            }
        }
    }

    //创建节点
    // 这里存在一个小bug，如果之前创建过了，那么重新加载的时候就不会创建了，那么这个永久节点就会一直存在
    // 也就意味着之前在zookeeper中的结果会影响着我们

    /*
    解决方案：
    如果存在的话，那么就将之前存在的值置空
    */
    public static void createNodeSimple(CuratorFramework client, String path) throws Exception {
        if (null == client.checkExists().forPath(path)) {
            client.create().creatingParentsIfNeeded().forPath(path);
        }else{
            setData("/cn/nhs/schedule/exec", new Instruct());
        }
    }

    private static void setData(String path, Object value) throws Exception {
        if (null == client.checkExists().forPath(path)) return;
        client.setData().forPath(path, JSON.toJSONString(value).getBytes(CHARSET_NAME));
    }


    //获取数据
    public static byte[] getData(CuratorFramework client, String path) throws Exception {
        return client.getData().forPath(path);
    }

    //添加临时节点数据
    public static void appendPersistentData(CuratorFramework client, String path, String data) throws Exception {
        PersistentEphemeralNode node = new PersistentEphemeralNode(client, PersistentEphemeralNode.Mode.EPHEMERAL, path, data.getBytes(Constants.Global.CHARSET_NAME));
        node.start();
        node.waitForInitialCreate(3, TimeUnit.SECONDS);
    }



}